package com.llx.tablesql


import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.scala._
import org.apache.flink.core.fs.FileSystem
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.table.api.scala.{BatchTableEnvironment, StreamTableEnvironment}
import org.apache.flink.table.api.{Table, TableEnvironment, Types}
import org.apache.flink.table.sinks.{CsvTableSink, TableSink}
import org.apache.flink.table.sources.{CsvTableSource, TableSource}
import org.apache.hadoop.yarn.webapp.example.HelloWorld.Hello

object TableSqlExample {


  case class Order(user: Long, product: String, amount: Int)
  /**getTableEnvironment:这个TableEnviromment是Table API和SQL集成的核心概念。它负责：
    * 在内部目录中注册一个表
    * 注册外部目录
    * 执行SQL查询
    * 注册用户定义的（标量，表格或聚合）函数
    * 转换DataStream或DataSet成Table
    * 持有一个ExecutionEnvironment或一个参考StreamExecutionEnvironment
    *
    * 1、在内部目录中注册一个表
    * TableEnvironment维护一个按名称注册的表的目录。有两种类型的表格，输入表格和输出表格。
    * 输入表可以在Table API和SQL查询中引用并提供输入数据。输出表可用于将表API或SQL查询的结果发送到外部系统
    * 输入表可以从各种来源注册：
    * 现有Table对象，通常是表API或SQL查询的结果。
    * TableSource，它访问外部数据，例如文件，数据库或消息传递系统。
    *  DataStream或DataSet来自DataStream或DataSet程序。
    * 输出表可以使用注册TableSink。
    *
    * batch query
    */
  def main(args: Array[String]): Unit = {


    /**
      * stream Query
      */
    val sEnv = StreamExecutionEnvironment.getExecutionEnvironment
    val sTableEnv = TableEnvironment.getTableEnvironment(sEnv)

    // set up execution environment
    val benv = ExecutionEnvironment.getExecutionEnvironment
    val btEnv = TableEnvironment.getTableEnvironment(benv)

    Table和DataStream和DataSet的集成(sTableEnv,sEnv)
    batch执行环境(btEnv,benv)
    将表转换为DataStream(sTableEnv,sEnv)
    将表转换为DataSet(btEnv,benv)
  }


  ///////////////////////////////////////////////////////////////////////////////////////////
  /**
    * 注册表
    */
  //  bTableEnv.registerTable("projectedTable", projTable )
  //查询
  //  val projTable: Table = bTableEnv.scan("projectedTable ").select("")

  /**
    * 2、注册一个tableSource
    *TableSource提供对存储在诸如数据库（MySQL，HBase等），具有特定编码（CSV，Apache [Parquet，Avro，ORC]，…）的文件的存储系统中的外部数据的访问或者消息传送系统（Apache Kafka，RabbitMQ，…）
    */
  //  val csvSource: TableSource = new CsvTableSource("/path/to/file")


  /**
    * 3、注册一个tableSink
    * 注册TableSink可用于将表API或SQL查询的结果发送到外部存储系统，如数据库，键值存储，消息队列或文件系统（使用不同的编码，例如CSV，Apache [Parquet ，Avro，ORC]，…）
    *
    */
  //  val csvSink: TableSink = new CsvTableSink("/path/to/file", ",")
  //  val fieldNames: Array[String] = Array("a", "b", "c")
  //  val fieldTypes: Array[TypeInformation[_]] = Array(Types.INT, Types.STRING, Types.LONG)
  //  bTableEnv.registerTableSink("CsvSinkTable", fieldNames, fieldTypes, csvSink)

///////////////////////////////////////////////////////////////////////////////////////////

  def batch执行环境(btEnv : BatchTableEnvironment, env: ExecutionEnvironment): Unit ={

    //加载外部数据
    val csvTableSource = CsvTableSource.builder()
      .path("data1.csv")//文件路径
      .field("id" , Types.INT)//第一列数据
      .field("name" , Types.STRING)//第二列数据
      .field("age" , Types.INT)//第三列数据
      .fieldDelimiter(",")//列分隔符，默认是"，"
      .lineDelimiter("\n")//换行符
      .ignoreFirstLine()//忽略第一行
      .ignoreParseErrors()//忽略解析错误
      .build()
    //将外部数据构建成表
    btEnv.registerTableSource("tableA" , csvTableSource)
    //TODO 1：使用table方式查询数据
    val table = btEnv.scan("tableA").select("id , name , age").filter("name == ‘lisi’")
    //将数据写出去(文件名字，分隔符，文件个数，写入规则)
    table.writeToSink(new CsvTableSink("bbb" , "," , 1 , FileSystem.WriteMode.OVERWRITE))
    //TODO 2：使用sql方式
     val sqlResult = btEnv.sqlQuery("select id,name,age from tableA where id > 0 order by id limit 2")
    //// //将数据写出去
     sqlResult.writeToSink(new CsvTableSink("aaaaaa.csv", ",", 1, FileSystem.WriteMode.OVERWRITE))
    env.execute()
  }

  ///////////////////////////////////////////////////////////////

  /**
    * Table和DataStream和DataSet的集成
    *
    * /**
    * * 在上面的例子讲解中，直接使用的是：registerTableSource注册表
    * * 对于flink来说，还有更灵活的方式：比如直接注册DataStream或者DataSet转换为一张表。
    * * 然后DataStream或者DataSet就相当于表，这样可以继续使用SQL来操作流或者批次的数据
    * * 语法：
    * * // get TableEnvironment
    * * // registration of a DataSet is equivalent
    * * Env:DataStream
    * * val tableEnv = TableEnvironment.getTableEnvironment(env)
    * *
    * * val stream: DataStream[(Long, String)] = …
    * *
    * * // register the DataStream as Table "myTable" with fields "f0", "f1"
    * * tableEnv.registerDataStream("myTable", stream)
    * *
    * *
    **/
    */
  def Table和DataStream和DataSet的集成(stEnv : StreamTableEnvironment, env: StreamExecutionEnvironment): Unit ={
    //构造数据
    val orderA: DataStream[Order] = env.fromCollection(Seq(
      Order(1L, "beer", 3),
      Order(1L, "diaper", 4),
      Order(3L, "rubber", 2)))
    val orderB: DataStream[Order] = env.fromCollection(Seq(
      Order(2L, "pen", 3),
      Order(2L, "rubber", 3),
      Order(4L, "beer", 1)))
    // 根据数据注册表
    stEnv.registerDataStream("OrderA", orderA)
    stEnv.registerDataStream("OrderB", orderB)
    // union the two tables
    val result = stEnv.sqlQuery(
      "SELECT * FROM OrderA WHERE amount > 2 UNION ALL " +
        "SELECT * FROM OrderB WHERE amount < 2")
    result.writeToSink(new CsvTableSink("ccc" , "," , 1 , FileSystem.WriteMode.OVERWRITE))
    env.execute()

  }


  /**
    * A Table可以转换成a DataStream或DataSet。通过这种方式，可以在Table API或SQL查询的结果上运行自定义的DataStream或DataSet程序
    * 1：将表转换为DataStream
    * 有两种模式可以将 Table转换为DataStream：
    * 1：Append Mode
    * 将一个表附加到流上
    * 2：Retract Mode
    * 将表转换为流
    * 语法格式：
    *
    * get TableEnvironment.
    * // registration of a DataSet is equivalent
    * // ge val tableEnv = TableEnvironment.getTableEnvironment(env)
    *
    * // Table with two fields (String name, Integer age)
    * val table: Table = …
    *
    *  // convert the Table into an append DataStream of Row
    * val dsRow: DataStream[Row]= tableEnv.toAppendStreamRow
    *
    *  // convert the Table into an append DataStream of Tuple2[String, Int]
    * val dsTuple: DataStream[(String, Int)] dsTuple=tableEnv.toAppendStream(String, Int)
    *
    *  // convert the Table into a retract DataStream of Row.
    *  // A retract stream of type X is a DataStream[(Boolean, X)].
    *  // The boolean field indicates the type of the change.
    *  // True is INSERT, false is DELETE
    *
    *  val retractStream: DataStream[(Boolean, Row)] = tableEnv.toRetractStreamRow
    */
  case class Peoject(user: Long, index: Int, content: String)
  def 将表转换为DataStream(stEnv : StreamTableEnvironment, env: StreamExecutionEnvironment): Unit ={
    val data = List(
      Peoject(1L, 1, "Hello"),
    Peoject(2L, 2, "Hello"),
    Peoject(3L, 3, "Hello"),
    Peoject(4L, 4, "Hello"),
    Peoject(5L, 5, "Hello"),
    Peoject(6L, 6, "Hello"),
    Peoject(7L, 7, "Hello World"),
    Peoject(8L, 8, "Hello World"),
    Peoject(8L, 8, "Hello World"),
    Peoject(20L, 20, "Hello World"))

    env.setParallelism(1)

    val stream = env.fromCollection(data)
    val table: Table = stEnv.fromDataStream(stream)
    //TODO 将table转换为DataStream----将一个表附加到流上Append Mode
    val appendStream: DataStream[Peoject] = stEnv.toAppendStream[Peoject](table)
    //TODO 将表转换为流Retract Mode true代表添加消息，false代表撤销消息
    val retractStream: DataStream[(Boolean, Peoject)] = stEnv.toRetractStream[Peoject](table)
    retractStream.print()
    env.execute()
}



  /**语法格式
    * // get TableEnvironment
    * // registration of a DataSet is equivalent
    * val tableEnv = TableEnvironment.getTableEnvironment(env)
    *
    * // Table with two fields (String name, Integer age)
    * val table: Table = …
    *
    * // convert the Table into a DataSet of Row
    * val dsRow: DataSet[Row] = tableEnv.toDataSetRow
    *
    * // convert the Table into a DataSet of Tuple2[String, Int]
    * val dsTuple: DataSet[(String, Int)] = tableEnv.toDataSet(String, Int)
    *
    */
  def 将表转换为DataSet(btEnv : BatchTableEnvironment, env: ExecutionEnvironment): Unit ={
    val data = List(
      Peoject(1L, 1, "Hello"),
      Peoject(2L, 2, "Hello"),
      Peoject(3L, 3, "Hello"),
      Peoject(4L, 4, "Hello"),
      Peoject(5L, 5, "Hello"),
      Peoject(6L, 6, "Hello"),
      Peoject(7L, 7, "Hello World"),
      Peoject(8L, 8, "Hello World"),
      Peoject(8L, 8, "Hello World"),
      Peoject(20L, 20, "Hello World"))
    //初始化环境，加载table数据

    env.setParallelism(1)

    val collection: DataSet[Peoject] = env.fromCollection(data)
    val table: Table = btEnv.fromDataSet(collection)
    //TODO 将table转换为dataSet
    val toDataSet: DataSet[Peoject] = btEnv.toDataSet[Peoject](table)
    toDataSet.print()
    env.execute()
  }

  /**
    * Mysql—canal—kafka—flink—hbase—web
    * Mysql的binlog日志作用是用来  记录mysql内部增删等   对mysql数据库有更新的内容的记录（对数据库的改动），
    * 对数据库的   查询select或show等不会被  binlog日志记录;   主要用于数据库的主从复制以及增量恢复。
    *
    * 开发：
      * 1、建表   mysql的binlog日志必须打开log-bin功能才能生存binlog日志
      *CREATE USER canal IDENTIFIED BY ‘canal’;
      * GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON . TO ‘canal’@’%’;
      * – GRANT ALL PRIVILEGES ON . TO ‘canal’@’%’ ;
      * FLUSH PRIVILEGES;
    *  1-1：
    *   修改/etc/my.cnf,在里面添加如下内容
    *     [mysqld]
          log-bin=/var/lib/mysql/mysql-bin 【binlog日志存放路径】
          binlog-format=ROW 【日志中会记录成每一行数据被修改的形式】
          server_id=1 【指定当前机器的服务ID（如果是集群，不能重复）】
    *   1-2：
    *     配置完毕之后，登录mysql，输入如下命令
    *     show variables like ‘%log_bin%’
    *
    * 2、安装cancel  编写cancel代码
    *     https://github.com/alibaba/canal/wiki/ClientExample
    *
    * 3、启动kafka并创建topic
    *
    */
}
